﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Data;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Imaging;
using System.Windows.Navigation;
using System.Windows.Shapes;

namespace RabbitMQ示例
{
    /// <summary>
    /// Hash功能处理
    /// </summary>
    public partial class ConsistentHashExchange示例 : UserControl
    {


        /// <summary>
        /// mq创建对像
        /// </summary>
        RabbitMQ.Client.ConnectionFactory faction = new RabbitMQ.Client.ConnectionFactory();

        /// <summary>
        /// 创建联接
        /// </summary>
        RabbitMQ.Client.IConnection conn = null;

        /// <summary>
        /// 初始化配置
        /// </summary>
        MqConfig conf = null;
        /// <summary>
        /// 联接模式字典
        /// </summary>
        private System.Collections.Concurrent.ConcurrentDictionary<string, MQChannelEntity> channelDic = new System.Collections.Concurrent.ConcurrentDictionary<string, MQChannelEntity>();
        /// <summary>
        /// 联接模式字典
        /// </summary>
        public ConcurrentDictionary<string, MQChannelEntity> ChannelDic
        {
            get
            {
                return channelDic;
            }

        }

        /// <summary>
        /// 
        /// </summary>
        public ConsistentHashExchange示例()
        {
            InitializeComponent();
        }

        /// <summary>
        /// 步骤1 初始化
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btn_Init_Click(object sender, RoutedEventArgs e)
        {

            RabbitMQConfig.Instance.Init();
            conf = RabbitMQConfig.Instance.Conf;


            faction.HostName = conf.HostName;
            faction.UserName = conf.UserName;
            faction.Password = conf.Password;
            faction.VirtualHost = conf.VirtualHost;
            //设置自动恢复功能
            faction.AutomaticRecoveryEnabled = true;
            faction.NetworkRecoveryInterval = new TimeSpan(0, 0, 5);
            faction.ContinuationTimeout = new TimeSpan(0, 0, 5);
            //创建联接对像
            conn = faction.CreateConnection(conf.HostName.Split(','));
            foreach (var item in conf.QueueCollection)
            {
                //创建或者设置频道
                RabbitMQ.Client.IModel channel = conn.CreateModel();
                //设置Exchange持久化
                channel.ExchangeDeclare(item.ExchangeName, item.ExchangeType, item.Durable);
                //Hash时不需要配置队列
                //创建队列 ,非排它,非自动删除

                channel.QueueDeclare(item.QueueName, item.Durable, false, false, null);

                //hash 算法不需要声明队列,它将自动分配
                if (item.ExchangeType != "x-consistent-hash")
                {
                    
                    channel.QueueBind(item.QueueName, item.ExchangeName, item.RoutingKey, null);
                }
                 //添加到集合中
                ChannelDic.TryAdd(item.RoutingKey, new MQChannelEntity() { Channel = channel, QueueInfo = item });
            }

            //声明Exchange及类型
            //channel.exchangeDeclare(EXCHANGE_NAME, "x-consistent-hash", true);
            //发送数据时
            //channel.basicPublish(EXCHANGE_NAME, message, null, message.getBytes());


            #region 打印状态线程

            WriteState = Task.Factory.StartNew(() =>
            {
                while (true)
                {

                    if (receiveCollection.Count > 0)
                    {
                        var ccc = receiveCollection.ToArray();
                        //打印所有队列状态
                        int count = ccc.Max(p => p.Value.ReceiveData.Count);
                        var arr = receiveCollection.ToArray();

                        for (int i = 0; i < count; i++)
                        {
                            List<string> strs = new List<string>();
                            foreach (var cc in arr)
                            {
                                string rf = "0000000000";
                                if (!cc.Value.ReceiveData.TryDequeue(out rf))
                                {
                                    rf = "          ";
                                }
                                strs.Add(string.Format("Q{0}:{1}", cc.Key, rf));
                            }

                            Console.WriteLine(string.Join(",", strs.ToArray()));
                        }

                    }
                    else
                    {
                        System.Threading.Thread.Sleep(100);
                    }
                }
            });

            #endregion

        }

        /// <summary>
        /// 发送线程
        /// </summary>
        Task SendTask = null;

        /// <summary>
        ///发送的记数
        /// </summary>
        long ina = 0;

        /// <summary>
        /// 发送循环
        /// </summary>
        bool SendCycle = true;


        /// <summary>
        ///  步骤2 循环发送数据
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btn_Send_Click(object sender, RoutedEventArgs e)
        {

            if (SendCycle)
            {
                SendCycle = false;
                System.Threading.Thread.Sleep(1000);
            }
            SendCycle = true;
            SendTask = Task.Factory.StartNew(() =>
            {
                Random dom = new Random();
                MQChannelEntity channel;
                if (ChannelDic.TryGetValue(ChannelDic.Keys.First(), out channel))
                {
                    ina = 0;
                    while (SendCycle)
                    {
                        if (channel.Channel.IsOpen)
                        {
                            System.Threading.Interlocked.Increment(ref ina);
                            int n = dom.Next(1, 9);
                            string message = n + ";Hello World " + Guid.NewGuid() + " " + ina;
                            var body = Encoding.UTF8.GetBytes(message);

                            string routeKey = "";
                            ////hash 算法
                            //if (channel.QueueInfo.ExchangeType.ToLower() == "x-consistent-hash")
                            //{
                            //    //数字终端号
                            //    routeKey = n.ToString();
                            //}
                            //else
                            //{
                                routeKey = string.Format(channel.QueueInfo.RoutingKey, n);
                            //}

                            //发布数据
                            channel.Channel.BasicPublish(channel.QueueInfo.ExchangeName, "abc" + routeKey, null, body);
                            System.Threading.Thread.Sleep(100);
                        }
                        else
                        {
                            return;
                        }
                    }

                }
            });
        }






        /// <summary>
        /// 接收信息集合
        /// </summary>
        ConcurrentDictionary<int, QueEnt> receiveCollection = new ConcurrentDictionary<int, QueEnt>();



        int sn = 0;

        /// <summary>
        /// 打印状态线程
        /// </summary>
        Task WriteState = null;

        /// <summary>
        ///  步骤3 增加一个接收器
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void btn_AddReceive_Click(object sender, RoutedEventArgs e)
        {





            Task.Factory.StartNew(() =>
         {
             Random dom = new Random();
             MQChannelEntity channela;





             if (ChannelDic.TryGetValue(ChannelDic.Keys.First(), out channela))
             {
                 sn++;
                 //临时多个队列对像
                 var quent = new QueEnt()
                 {
                     CanRun = true,
                     QueIndex = sn,
                     ReceiveData = new ConcurrentQueue<string>(),
                 };

                 #region 创建多个Channel用于接收数据

                 var item = channela.QueueInfo;
                 //创建或者设置频道
                 RabbitMQ.Client.IModel channel = conn.CreateModel();
                 //设置Exchange持久化
                 channel.ExchangeDeclare(item.ExchangeName, item.ExchangeType, item.Durable);
                 //创建队列 ,非排它,非自动删除


                 //权重, 指定 队列名 ,用于故障恢复时可用,用数字表示
                 string BindRouteKey = quent.QueIndex.ToString();
                 //队列名前缀
                 String queueName = "ConsistentHash"+BindRouteKey;

                 //固定队列名
                 channel.QueueDeclare(queueName, item.Durable, false, false, null);

                 channel.QueueBind(queueName, item.ExchangeName, BindRouteKey, null);

                 //队列索引
                 receiveCollection.TryAdd(quent.QueIndex, quent);

                 #endregion

                 ina = 0;
                 //接收队像
                 RabbitMQ.Client.QueueingBasicConsumer queue = new RabbitMQ.Client.QueueingBasicConsumer(channel);
                 // var queuea = channel.Channel.QueueDeclare();
                 channel.BasicConsume("", true, queue);


                 while (queue.IsRunning || channel.IsOpen)
                 {
                     try
                     {
                         var q = queue.Queue.Dequeue();
                         if (q != null)
                         {
                             string str = System.Text.UTF8Encoding.UTF8.GetString(q.Body);
                             //quent.ReceiveData.Enqueue(long.Parse(str.Split(';')[0]).ToString("d10") + "," + str);
                             quent.ReceiveData.Enqueue(long.Parse(str.Split(';')[0]).ToString("d10"));
                         }
                     }
                     catch (Exception ex)
                     {
                         if (!string.IsNullOrEmpty(ex.Message))
                         {
                             if (ex.Message.Contains("SharedQueue closed"))
                             {
                                 Console.WriteLine("联接断开,或者队列移除");
                             }
                         }
                         else
                         {
                             Console.WriteLine("队列名:{0},读取时出现问题", queueName);
                         }

                         return;

                     }

                 }

             }
         });
        }

        /// <summary>
        /// 减少一个队列后数据重新分配
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void button_Click(object sender, RoutedEventArgs e)
        {
            减少一个队列();
        }

        /// <summary>
        /// 减少一个队列后数据重新分配
        /// </summary>
        public void 减少一个队列()
        {
            Random dom = new Random();
            MQChannelEntity channela;





            if (ChannelDic.TryGetValue(ChannelDic.Keys.First(), out channela))
            {
                if (receiveCollection.Count > 0)
                {
                    var item = receiveCollection.FirstOrDefault();

                    string queu = "ConsistentHash" + item.Key.ToString();

                    //解除邦定
                    channela.Channel.QueueUnbind(queu, channela.QueueInfo.ExchangeName, item.Key.ToString(), null);

                    //删除指定队列
                    //channela.Channel.QueueDelete(queu);

                    Console.WriteLine("-----------减少队列:{0},数据重新分配", item.Key);

                    var delobj = item.Value;

                    receiveCollection.TryRemove(item.Key, out delobj);
                }
            }
        }


        /// <summary>
        /// 队列对像实体
        /// </summary>
        public class QueEnt
        {
            /// <summary>
            /// 队列索引
            /// </summary>
            public int QueIndex { get; set; }

            /// <summary>
            /// 是否可以运行
            /// </summary>
            public bool CanRun { get; set; }

            /// <summary>
            /// 接收到的数据队列
            /// </summary>
            public System.Collections.Concurrent.ConcurrentQueue<string> ReceiveData { get; set; }

        }


    }
}

